package com.seaboxdata.kafka;

import java.util.Properties;
import java.util.concurrent.Future;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

/**
 * @author
 * @create 2020-08-28 10:09
 **/
@Component
@Slf4j
@Configuration
public class BaseStringKafkaProducer {

    private KafkaProducer<String, String> producer;

    private Properties props;

    private static boolean isWin = System.getProperty("os.name").toLowerCase().contains("win");

    @Value("${kafka.server}")
    private String kafkaServer;

    @Value("${kafka.jaas.config}")
    private String authConfig;

    @Value("${krb5.config}")
    private String krb5Config;

    @Value("${kafka.win.jaas.config}")
    private String authWinConfig;

    @Value("${krb5.win.config}")
    private String krb5WinConfig;

    @Value("${kafka.kerberos.enable}")
    private String isKerberos;

    public BaseStringKafkaProducer() {
        props = new Properties();
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 1000);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
    }

    public void sendMessageToKafka(String msg, String topic) throws Exception {
        if ("true".equals(isKerberos)) {
            if (isWin) {
                //如果是windows指定本地的配置文件,本地调试使用
                this.authConfig = authWinConfig;
                this.krb5Config = krb5WinConfig;
            }
            System.setProperty("java.security.auth.login.config", authConfig);
            System.setProperty("java.security.krb5.conf", krb5Config);
            props.put("security.protocol", "SASL_PLAINTEXT");
            props.put("sasl.kerberos.service.name", "kafka");
            props.put("sasl.mechanism", "GSSAPI");
        }

        if (producer == null) {
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
            producer = new KafkaProducer<String, String>(props);
        }
        Future<RecordMetadata> send = producer.send(new ProducerRecord<String, String>(topic, msg));
    }
}
